package com.rabbitmq.utils;

import com.rabbitmq.dynamic.Receive;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author: TongRui乀
 * @Date: 2020-07-24 9:34
 * @description：
 */
@Component
public class RabbitDynamicUtil {

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private SimpleMessageListenerContainer simpleMessageListenerContainer;

    /**
     *  创建一个Queue
     * @param queueName  队列名称
     */
    public Queue createQueue(String queueName){
        Queue queue = QueueBuilder.durable(queueName).build();
        rabbitAdmin.declareQueue(queue);
        return queue;
    }

    /**
     *  创建一个Exchange
     * @param exchangeName 交换机名称
     */
    public Exchange createDelayedExchange(String exchangeName){
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        CustomExchange exchange = new CustomExchange(exchangeName, "x-delayed-message", true, false, args);
        rabbitAdmin.declareExchange(exchange);
        return exchange;
    }

    /**
     *  绑定 Queue 和 Exchange
     * @param queue     队列
     * @param exchange  交换机
     * @param bindingKey 两者的绑定key
     */
    public void createBinding(Queue queue, Exchange exchange, String bindingKey){

        Binding binding = BindingBuilder.bind(queue).to(exchange).with(bindingKey).noargs();

        rabbitAdmin.declareBinding(binding);
    }


    /**
     *  删除队列
     * @param queueName  队列的名字
     * @param unUsed  true 标识在没有被使用的时候删除
     * @param empty  true 表示在队列为空的时候删除
     */
    public void deleteQueue(String queueName, boolean unUsed, boolean empty){

        rabbitAdmin.deleteQueue(queueName, unUsed, empty);

    }

    /**
     *  删除队列
     * @param queueName  队列的名字
     */
    public void deleteQueue(String queueName){

        rabbitAdmin.deleteQueue(queueName);

    }

    /**
     *  删除exchange
     * @param exchangeName
     */
    public void deleteExchange(String exchangeName){
        rabbitAdmin.deleteExchange(exchangeName);
    }



    /**
     * 从指定队列中消费消息
     * @param queueName
     */
    public String receiveFromQueue(String queueName){

        byte[] res = (byte[]) rabbitTemplate.receiveAndConvert(queueName);

        return res == null ? "" : new String(res);
    }


    /**
     *  直接发送消息到队列
     * @param exchangeName
     * @param queueName
     * @param message
     */
    public void sendToQueue(String exchangeName, String queueName, String message){
        //设置消息内容的类型，默认是 application/octet-stream 会是 ASCII 码值
        rabbitTemplate.convertAndSend(exchangeName, queueName, message);
    }

    /**
     *  通过bindingKey 发送消息
     * @param exchangeName
     * @param routingKey
     * @param message
     */
    public void sendMessageToExchange(String exchangeName, String routingKey, String message){
        rabbitTemplate.send(exchangeName, routingKey, buildMessage(MessageProperties.CONTENT_TYPE_JSON,message, null));
    }


    /**
     *  通过bindingKey 发送延时消息
     * @param exchangeName
     * @param routingKey
     * @param message
     * @param delayed
     */
    public void sendDelayedMessageToExchange(String exchangeName, String routingKey, String message, Integer delayed){
        rabbitTemplate.send(exchangeName, routingKey, buildMessage(MessageProperties.CONTENT_TYPE_JSON,message, delayed));
    }


    /**
     *   创建带请求头的消息
     * @param messageType
     * @param msg
     * @param delayed
     * @return
     */
    public Message buildMessage(String messageType, String msg, Integer delayed){
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setContentType(messageType);
        messageProperties.setDelay(delayed);
        return new Message(msg.getBytes(),messageProperties);
    }

    public void addListener(String queueName){
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new Receive());
        messageListenerAdapter.addQueueOrTagToMethodName(queueName,"handleMessage");
        simpleMessageListenerContainer.addQueueNames(queueName);
        simpleMessageListenerContainer.setMessageListener(messageListenerAdapter);
    }

}
